package com.its.common.disruptor.consumer;

import com.its.common.disruptor.dto.TranslatorDataWapper;
import lombok.extern.slf4j.Slf4j;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Author xiaxp
 * @Date 2021-11-25 16:43
 * @Description 定时定量处理
 */
@Slf4j
public abstract class TimedAndMaxCountOnceMessageConsumer extends BaseMessageConsumer {
    private final ReentrantLock LOCK = new ReentrantLock();

    private final Condition CONDITION = LOCK.newCondition();
    private InheritableThreadLocal<LongAdder> contiNoDataCount = new InheritableThreadLocal<>();

    private InheritableThreadLocal<LongAdder> count = new InheritableThreadLocal<>();
    private InheritableThreadLocal<Long> start = new InheritableThreadLocal<>();


    private InheritableThreadLocal<ConcurrentLinkedQueue> datasets = new InheritableThreadLocal<>();
    private InheritableThreadLocal<Integer> maxCount = new InheritableThreadLocal<>();
    private InheritableThreadLocal<Long> ms = new InheritableThreadLocal<>();;

    public TimedAndMaxCountOnceMessageConsumer(String consumerId, int maxCount, long ms) {
        super(consumerId);
        count.set(new LongAdder());
        contiNoDataCount.set(new LongAdder());
        start.set(System.currentTimeMillis());
        datasets.set(new ConcurrentLinkedQueue<TranslatorDataWapper>());
        this.maxCount.set(maxCount);
        this.ms.set(ms);
        if(ms < 0){
            ms = 1999999999;
        }
        this.startTimer();
    }

    @Override
    public void onEvent(TranslatorDataWapper event) throws Exception {
        this.LOCK.lock();
        try {
            datasets.get().offer(event);
            count.get().increment();

            contiNoDataCount.get().reset();
            CONDITION.signal();
        }catch (Exception e){
            System.out.println(e);
        }finally {
            this.LOCK.unlock();
        }
    }

    private void startTimer(){
        ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
        service.scheduleAtFixedRate(() -> {
            try {
                if(count.get().sum() >= maxCount.get() || System.currentTimeMillis() - start.get() >= ms.get()){
                    moveDatasetToHandlerAndReset();
                }
            }catch (Exception e){
                System.out.println(e);
            }
            if(count.get().sum() == 0){
                if(contiNoDataCount.get().sum() > 30) {
                    try {
                        this.LOCK.lock();
                        CONDITION.await();
                    } catch (InterruptedException e) {
                        System.out.println(e);
                    } finally {
                        this.LOCK.unlock();
                    }
                }else{
                    contiNoDataCount.get().increment();
                }
            }
        }, 1,1, TimeUnit.SECONDS);
    }

    private void moveDatasetToHandlerAndReset(){
        List<TranslatorDataWapper> destList = new ArrayList<>();
        try {
            ConcurrentLinkedQueue queue = datasets.get();
            for (int i = 0; i < maxCount.get(); i++) {
                TranslatorDataWapper wapper = (TranslatorDataWapper) queue.poll();
                if(null != wapper){
                    destList.add(wapper);
                }else{
                    break;
                }
            }
            log.info(Thread.currentThread().getName() + " - " +destList.size());
            count.get().reset();
            start.set(System.currentTimeMillis());
        } finally {
            this.handleDatasets(destList);
        }
    }

    /** 继承实现数据集合的处理逻辑
     * @param list
     */
    public abstract void handleDatasets(List<TranslatorDataWapper> list);

    @Override
    public void handleData(TranslatorDataWapper translatorDataWapper) {}
}
